package com.stock;

import java.io.*;
import java.util.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.fs.Path;
import java.net.URI;

public class SentimentWordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    private static final IntWritable one = new IntWritable(1);
    private Text sentimentWord = new Text();
    private Set<String> stopWords = new HashSet<>();
    
    // 常见停用词集合
    private static final Set<String> DEFAULT_STOP_WORDS = new HashSet<>(Arrays.asList(
        "a", "an", "the", "and", "or", "but", "in", "on", "at", "to", "for", "of", 
        "with", "by", "from", "up", "about", "into", "through", "during", "before",
        "after", "above", "below", "between", "among", "is", "are", "was", "were",
        "be", "been", "being", "have", "has", "had", "do", "does", "did", "will",
        "would", "could", "should", "may", "might", "must", "can", "it", "its",
        "they", "them", "their", "this", "that", "these", "those", "i", "you", "he",
        "she", "we", "us", "our", "my", "your", "his", "her", "me", "him", "user"
    ));

    @Override
    protected void setup(Context context) throws IOException {
        // 先添加默认停用词
        stopWords.addAll(DEFAULT_STOP_WORDS);
        
        // 再从文件加载额外的停用词
        URI[] stopWordsFiles = context.getCacheFiles();
        if (stopWordsFiles != null && stopWordsFiles.length > 0) {
            for (URI stopFile : stopWordsFiles) {
                Path path = new Path(stopFile.getPath());
                try (BufferedReader br = new BufferedReader(new FileReader(path.getName()))) {
                    String line;
                    while ((line = br.readLine()) != null) {
                        stopWords.add(line.trim().toLowerCase());
                    }
                }
            }
        }
    }

    @Override
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        String line = value.toString().trim();
        
        // 跳过表头
        if (line.startsWith("Text,Sentiment") || line.isEmpty()) return;

        // 使用改进的CSV解析
        String[] parts = parseCSVLine(line);
        if (parts.length < 2) return;

        String text = parts[0];
        String sentiment = parts[1].trim();

        // 验证情感标签
        if (!sentiment.equals("1") && !sentiment.equals("-1")) {
            return;
        }

        // 分词处理
        List<String> words = tokenizeText(text);
        for (String word : words) {
            if (isValidWord(word)) {
                sentimentWord.set(sentiment + "\t" + word);
                context.write(sentimentWord, one);
            }
        }
    }

    /**
     * 改进的CSV解析方法，正确处理引号内的逗号
     */
    private String[] parseCSVLine(String line) {
        List<String> result = new ArrayList<>();
        boolean inQuotes = false;
        StringBuilder field = new StringBuilder();
        
        for (int i = 0; i < line.length(); i++) {
            char c = line.charAt(i);
            
            if (c == '"') {
                inQuotes = !inQuotes;
            } else if (c == ',' && !inQuotes) {
                result.add(field.toString().replaceAll("\"", "").trim());
                field.setLength(0);
            } else {
                field.append(c);
            }
        }
        // 添加最后一个字段
        result.add(field.toString().replaceAll("\"", "").trim());
        
        return result.toArray(new String[0]);
    }

    /**
     * 改进的分词方法
     */
    private List<String> tokenizeText(String text) {
        List<String> tokens = new ArrayList<>();
        
        if (text == null || text.isEmpty()) {
            return tokens;
        }
        
        // 转换为小写
        text = text.toLowerCase();
        
        // 移除URL、特殊字符，但保留字母、数字和基本标点
        text = text.replaceAll("https?://\\S+", ""); // 移除URL
        text = text.replaceAll("[^a-zA-Z0-9\\s',.!?-]", " "); // 保留字母数字和基本标点
        
        // 分词
        String[] words = text.split("\\s+");
        
        for (String word : words) {
            if (word == null || word.isEmpty()) continue;
            
            // 清理单词：移除开头和结尾的非字母字符
            String cleaned = word.replaceAll("^[^a-zA-Z]+|[^a-zA-Z]+$", "");
            
            if (!cleaned.isEmpty()) {
                tokens.add(cleaned);
            }
        }
        
        return tokens;
    }

    /**
     * 验证单词是否有效
     */
    private boolean isValidWord(String word) {
        if (word == null || word.isEmpty()) return false;
        
        // 长度检查
        if (word.length() <= 1) return false;
        
        // 停用词检查
        if (stopWords.contains(word)) return false;
        
        // 纯数字检查
        if (word.matches("\\d+")) return false;
        
        // 检查是否包含至少一个字母
        if (!word.matches(".*[a-zA-Z].*")) return false;
        
        // 检查是否包含太多重复字符（如"aaaa"）
        if (hasRepeatingChars(word, 3)) return false;
        
        return true;
    }
    
    /**
     * 检查单词是否包含太多重复字符
     */
    private boolean hasRepeatingChars(String word, int maxRepeat) {
        if (word.length() < maxRepeat + 1) return false;
        
        for (int i = 0; i <= word.length() - maxRepeat; i++) {
            char c = word.charAt(i);
            boolean allSame = true;
            for (int j = 1; j < maxRepeat; j++) {
                if (word.charAt(i + j) != c) {
                    allSame = false;
                    break;
                }
            }
            if (allSame) return true;
        }
        return false;
    }
}